package com.pocket.insight.base.sql.dataset;

import com.pocket.insight.base.context.SpringContext;
import com.pocket.insight.base.exception.BizException;
import com.pocket.insight.base.sql.dataset.resource.BaseResourceCreator;
import com.pocket.insight.base.sql.dataset.resource.NativeResourceCreator;
import com.pocket.insight.entity.Dataset;
import com.pocket.insight.service.DatasetService;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.tuple.Pair;

import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 描述：数据集工厂
 *
 * @author shuize
 * @create 2024/12/4
 */
public class DatasetResourceFactory {

    protected static DatasetService datasetService = SpringContext.getBean(DatasetService.class);

    // 创建数据集资源类
    public static DatasetResource create(Dataset dataset) {
        return create(dataset, false, null);
    }
    public static DatasetResource create(Dataset dataset, RuntimeData runtimeData) {
        return create(dataset, false, runtimeData);
    }
    public static DatasetResource create(Dataset dataset, Boolean forceNative, RuntimeData runtimeData) {
        return createAndInitWithResourceMapCache(dataset, new AtomicBoolean(forceNative), runtimeData);
    }

    //
    private static DatasetResource createAndInitWithResourceMapCache(Dataset dataset, AtomicBoolean forceNative, RuntimeData runtimeData) {
        Object resourceMap;
        if (runtimeData == null){
            runtimeData = new RuntimeData();
            resourceMap = new HashMap();
            runtimeData.setShareResourceMap((Map)resourceMap);
            runtimeData.setShareWithMap(new HashMap());
        } else {
            resourceMap = runtimeData.getShareResourceMap();
            Map<String, Pair<String, String>> withMap = runtimeData.getShareWithMap();
            if (resourceMap != null && withMap == null){
                throw new IllegalArgumentException("withMap cannot be null when resourceMap is not null");
            }
            if (resourceMap == null) {
                resourceMap = new HashMap();
                runtimeData.setShareResourceMap((Map)resourceMap);
                if (withMap == null) {
                    runtimeData.setShareWithMap(new HashMap());
                }
            }
        }
        dataset = datasetService.getDatasetWithAuth(dataset);
        BaseResourceCreator creator = dispatchResource(dataset, forceNative, runtimeData);
        creator.init();
        return creator.getDatasetResource();
    }

    private static BaseResourceCreator dispatchResource(Dataset dataset, AtomicBoolean forceNative, RuntimeData runtimeData) {
        Map<Long, DatasetResource> resourceMap = runtimeData.getShareResourceMap();
        AtomicBoolean forceNativeForSub = new AtomicBoolean(forceNative.get());
        Dataset.Options options = dataset.getOptions();
        List<Long> directDependantDatasetList = new ArrayList();
        Long appIdOfDataset = DatasetUtil.getAppDatasetIdFromId(dataset.getAppId(), dataset.getId()).getLeft();


        BiConsumer<Long, Long> collectDataset = (appId, datasetId) -> {
            if (!resourceMap.containsKey(datasetId)) {
                RuntimeData aimRuntimeData = runtimeData
                        .copyWithoutApplicableDatasetAcl()
                        .setApplyDatasetAcl(BooleanUtils.isTrue(dataset.getEnableUpstreamAcl()))
                        .setApplyParentAcl(!DatasetUtil.getAppDatasetIdFromId(appId, datasetId).getLeft().equals(appIdOfDataset));
                resourceMap.put(datasetId, createAndInitWithResourceMapCache(datasetService.getById(appId, datasetId), forceNativeForSub, aimRuntimeData));
            }
            if (!directDependantDatasetList.contains(datasetId)) {
                directDependantDatasetList.add(datasetId);
            }
        };

        boolean isViewerMode = isViewerMode(dataset.getAppId());
        BiFunction<Function<Void, Void>, Function<Long, BaseResourceCreator>, BaseResourceCreator> tryFallback = (resourceCollector, resourceCreator) -> {
            for (int i = 0; i < 2; i++) {
                resourceCollector.apply(null);
                List<Long> effectiveStorageConnectionIdList = directDependantDatasetList.stream()
                                                                .filter(item -> !resourceMap.get(item).getDataset().getCanPushToMemory())
                                                                .map(item -> resourceMap.get(item).getEffectiveStorageConnectionId())
                                                                .filter(Objects::nonNull)
                                                                .distinct()
                                                                .collect(Collectors.toList());
                if (CollectionUtils.isEmpty(effectiveStorageConnectionIdList)) {
                    throw new BizException("ERROR_BAD_PARAMETER：" + effectiveStorageConnectionIdList);
                }
                if (effectiveStorageConnectionIdList.size() == 1) {
                    if (resourceCreator == null) {
                        return null;
                    }
                    BaseResourceCreator result = resourceCreator.apply(effectiveStorageConnectionIdList.get(0));
                    if (result != null) {
                        return result;
                    }
                }
                if (i == 0) {
                    forceNativeForSub.set(true);
                    Iterator datasetIterator = directDependantDatasetList.iterator();
                    while(datasetIterator.hasNext()) {
                        resourceMap.remove((Long)datasetIterator.next());
                    }
                }
            }
            return null;
        };

        List<Long> datasetsInFilter = DatasetUtil
                .fetchUpstreamDatasets(options, dataset.getAppId(), dataset.getId())
                .stream().map(s -> s.getSourceId()).distinct().filter(item -> !item.equals(dataset.getId())).collect(Collectors.toList());
        if (datasetsInFilter.size() > 1) {
            datasetService.batchFetchDatasetsInfo(dataset.getAppId(), datasetsInFilter, false);
        }

        try {
            BaseResourceCreator result;
            Function<Void, Void> collector;
            Function<Long, BaseResourceCreator> creator;
            switch (options.fetchType()) {
                case connection:
                case internal_storage:
                    Long myEffectiveStorageConnectionId = dataset.getOptions().getConnectionId();
                    if (CollectionUtils.isEmpty(datasetsInFilter)){
                        return new NativeResourceCreator(dataset,  runtimeData.setShareResourceMap(null))
                                .setEffectiveStorageConnectionId(myEffectiveStorageConnectionId)
                                .setDatasetForceNativeForSub(forceNativeForSub.get());
                    }
                    collector = (aVoid) -> {
                        datasetsInFilter.forEach( item -> collectDataset.accept(dataset.getAppId(), item));
                        return null;
                    };
                    creator = (effectiveStorageConnectionId)
                            -> new NativeResourceCreator(dataset, runtimeData.setShareResourceMap(null))
                            .setDatasetForceNativeForSub(forceNativeForSub.get())
                            .setEffectiveStorageConnectionId(myEffectiveStorageConnectionId);
                    result = tryFallback.apply(collector, creator);
                    if (result == null) {
                        throw new BizException("ERROR_UNSUPPORTED_CROSS_CONNECTION_FUSION");
                    }
                    return result;
                case fusion:
                case union:
                case aggregate:
                case reference:
                case extend:
                case pivot:
                case unpivot:
            }
        } finally {
            //DatasetUtil.setTopDatasetId(topDataset);
        }
        throw new BizException("ERROR_SQL_QUERY_TYPE_UNSUPPORTED");
    }


    private static boolean isViewerMode(Long appId) {
        return false;
    }
}
